package com.wudl.core;

import org.apache.flink.shaded.netty4.io.netty.handler.codec.http2.Http2Exception;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.Arrays;
/**
 * TODO
 * <p>
 * 将数据流进行后，如何从流中将不同的标记取出来，这时就需要用select 算子
 *
 * @author wudl
 * @version 1.0
 * @date 2020/12/23 15:29
 */
public class TransformSelect {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Integer> inpputDs = env.fromCollection(Arrays.asList(1, 2, 4, 5, 7, 8, 9));
        SplitStream<Integer> splitStream = inpputDs.split(new OutputSelector<Integer>() {
            @Override
            public Iterable<String> select(Integer value) {
                if (value <= 3) {
                    return Arrays.asList("3");
                } else if (value > 3 || value < 7) {
                    return Arrays.asList("4");
                } else {
                    return Arrays.asList("10");
                }
            }
        });
/**
 * 通过不同的标签取出不同的流
 *一个流对应的标签名称，取出来的时候，给定名称就行
 */
        splitStream.select("3").print("3");
        splitStream.select("4").print("4");
        splitStream.select("10").print("10");
        env.execute();

    }
}
